package com.intct.utils;

import com.alibaba.fastjson.JSONObject;
import com.intct.common.Constant;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;
import java.util.Map;

/**
 * @author gufg
 * @since 2025-09-18 16:25
 */
public class HBaseUtil {

    public static Connection connection;

    static {
        try {
            connection = ConnectionFactory.createConnection();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 创建HBase异步连接
     * @return
     * @throws Exception
     */
    public static AsyncConnection getAsyncConnection() throws Exception {
        return ConnectionFactory.createAsyncConnection().get();
    }

    /**
     * 关闭HBase异步连接
     * @param asyncConnection
     * @throws IOException
     */
    public static void closeAsycnConnection(AsyncConnection asyncConnection) throws IOException {
        if (asyncConnection != null) {
            asyncConnection.close();
        }
    }

    public static void createTable(String ns, String tName, String familyName) throws IOException {
        // 创建admin
        Admin admin = connection.getAdmin();

        // TableName
        TableName tableName = TableName.valueOf(ns, tName);

        try {
            // 判断表不存
            if (!admin.tableExists(tableName)) {
                TableDescriptorBuilder tableDescriptorBuilder = TableDescriptorBuilder.newBuilder(tableName);

                ColumnFamilyDescriptorBuilder columnFamilyDescriptorBuilder = ColumnFamilyDescriptorBuilder.newBuilder(familyName.getBytes());

                columnFamilyDescriptorBuilder.setMaxVersions(3);

                tableDescriptorBuilder.setColumnFamily(columnFamilyDescriptorBuilder.build());


                admin.createTable(tableDescriptorBuilder.build());
            }
        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            // 关闭admin
            if (admin != null) {
                admin.close();
            }
        }
    }

    public static void createTable(String ns, String tName, String familyName, byte[][] splitKeys) throws IOException {
        // 创建admin
        Admin admin = connection.getAdmin();

        // TableName
        TableName tableName = TableName.valueOf(ns, tName);

        try {
            // 判断表不存
            if (!admin.tableExists(tableName)) {
                TableDescriptorBuilder tableDescriptorBuilder = TableDescriptorBuilder.newBuilder(tableName);

                ColumnFamilyDescriptorBuilder columnFamilyDescriptorBuilder = ColumnFamilyDescriptorBuilder.newBuilder(familyName.getBytes());

                columnFamilyDescriptorBuilder.setMaxVersions(3);

                tableDescriptorBuilder.setColumnFamily(columnFamilyDescriptorBuilder.build());


                admin.createTable(tableDescriptorBuilder.build(), splitKeys);
            }
        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            // 关闭admin
            if (admin != null) {
                admin.close();
            }
        }
    }
    public static void put(String ns, String tName, JSONObject jsonObj) throws IOException {
        String familyName = jsonObj.getString("sink_family");
        String rowkey = jsonObj.getString("rowkey");
        JSONObject dataObj = jsonObj.getJSONObject(Constant.FLINK_CDC_MYSQL_JSON_AFTER_KEY);

        // 创建table
        // TableName
        TableName tableName = TableName.valueOf(ns, tName);

        Table table = connection.getTable(tableName);

        try {
            Put put = new Put(rowkey.getBytes());

            for (Map.Entry<String, Object> entry : dataObj.entrySet()) {
                put.addColumn(familyName.getBytes(), entry.getKey().getBytes(), String.valueOf(entry.getValue()).getBytes());
            }

            // 插入数据
            table.put(put);
        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            if (table != null) {
                table.close();
            }
        }
    }

    /**
     *
     * @param ns
     * @param tName
     * @param rowkey
     * @return
     * @throws Exception
     */
    public static JSONObject getRowkeyData(AsyncConnection asyncConn, String ns, String tName, String rowkey) {
        try {
            TableName tableNameObj = TableName.valueOf(ns, tName);
            AsyncTable<AdvancedScanResultConsumer> asyncTable = asyncConn.getTable(tableNameObj);
            Get get = new Get(Bytes.toBytes(rowkey));
            Result result = asyncTable.get(get).get();
            Cell[] cells = result.rawCells();
            if (cells != null) {
                JSONObject jsonObj = new JSONObject();
                for (Cell cell : cells) {
                    String columnName = Bytes.toString(CellUtil.cloneQualifier(cell));
                    String columnValue = Bytes.toString(CellUtil.cloneValue(cell));
                    jsonObj.put(columnName, columnValue);
                }
                return jsonObj;
            }
        } catch (Exception e) {
            //throw new RuntimeException(e);
        }
        return null;
    }

}
